package com.cloxtec.scada.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.cloxtec.scada.utils.LogHelper;
import com.cloxtec.scada.utils.ParamHelper;
import com.cloxtec.scada.utils.UUIDGenerator;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;


@Component
@Slf4j
public class MqttPubClient {

    private Timer heartbeatTimer;
    private MqttAsyncClient mqttClient;
    private MqttConnectOptions connOpts;
    private MemoryPersistence persistence;
    private IMqttToken tok;

    boolean bRunStatus;
    private JSONObject mqttParam;

    private String userName;
    private String userPwd;
    private String serverURI;
    private JSONArray subTopics;
    private String clientId;
    private int subQos;

    @Resource
    private ParamHelper paramHelper;

    @Resource
    private LogHelper logHelper;


    public MqttPubClient() {
        bRunStatus = false;
//        heartbeatTimer = new Timer();
//        heartbeatTimer.schedule(new TimerTask() {
//            @Override
//            public void run() {
//                try {
//                    broadcastOnlineMsg();
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
//            }
//        }, 5000, 5000);//5秒后开始， 1 分钟启动一次，刷新设备在线状态
    }

    public void start() {
//        mqttParam = paramHelper.getParamById("param_mqtt", "mqtt_server");
        JSONObject mockMqttParam = new JSONObject();
        mockMqttParam.put("client_id", "default_client_id"); // 使用默认client_id或你想要的任何值
        mockMqttParam.put("url", "tcp://127.0.0.1:1883"); // 其他必要的MQTT参数
        mockMqttParam.put("server_url", "tcp://127.0.0.1:1883"); // 其他必要的MQTT参数
        mockMqttParam.put("username", "admin"); // 其他必要的MQTT参数
        mockMqttParam.put("password", "123456"); // 其他必要的MQTT参数
        // 使用模拟的MQTT参数
        mqttParam = mockMqttParam;


        String id = mqttParam.getString("client_id");
        id += UUIDGenerator.getUUID32();
        mqttParam.put("client_id", id);
        log.info("mqtt param:" + mqttParam.toJSONString());

        JSONObject jsonErr = new JSONObject();
        if (mqttParam != null) {
            initParam(mqttParam);
            connect(jsonErr);
            log.info("gateway mqtt client start: " + jsonErr.getString("err"));
            logHelper.saveLog("gateway mqtt client start: " + jsonErr.getString("err"));
        } else {
            log.info("gateway mqtt client init failed");
            logHelper.saveLog("gateway mqtt client init failed");
        }
    }

    public void stop() throws MqttException {
        if (mqttClient != null) {
            mqttClient.disconnectForcibly();
            mqttClient.close();
        }

        onStop();
        //broadcastOperationMsg("MQTT服务器已经断开连接...");
    }

    public void onStart() {
        bRunStatus = true;
    }

    public void onStop() {
        bRunStatus = false;
    }

    public void restart() throws MqttException {
        if (bRunStatus) {
            stop();
        }

        start();
    }

    public boolean getStatus() {
        return bRunStatus;
    }

    public void subscribe(String topic) throws MqttException {
        mqttClient.subscribe(topic, 0);
    }

    public void unSubscribe(String topic) throws MqttException {
        mqttClient.unsubscribe(topic);
    }

    public void initParam(JSONObject param) {
        userName = param.getString("username");
        userPwd = param.getString("password");
        serverURI = param.getString("url");
        subTopics = param.getJSONArray("sub_topics");
        clientId = param.getString("client_id");
        log.info("clientId:" + clientId);
    }

    public void connect(JSONObject jsonErr) {
        try {
            persistence = new MemoryPersistence();
            String strId = clientId;
            if (clientId.length() == 0) {
                strId = UUIDGenerator.getUUID32();
            }

            stop();

            mqttClient = new MqttAsyncClient(serverURI, strId, persistence);

            mqttClient.setCallback(new MqttCallbackExtended() {

                @Override
                public void connectionLost(Throwable throwable) {
                    connectionLostCallback(throwable);
                }

                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    messageArrivedCallback(s, mqttMessage);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    deliveryCompleteCallback(iMqttDeliveryToken);
                }

                @Override
                public void connectComplete(boolean b, String s) {
                    try {
                        connectCompleteCallback(b, s);
                    } catch (MqttException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });

            connOpts = new MqttConnectOptions();
            if (userName.length() > 0) {
                connOpts.setUserName(userName);
            }

            if (userPwd.length() > 0) {
                connOpts.setPassword(userPwd.toCharArray());
            }

            connOpts.setCleanSession(true);
            connOpts.setKeepAliveInterval(90);
            connOpts.setAutomaticReconnect(true);
            JSONObject jsonWill = new JSONObject();
            jsonWill.put("webMsgType", 1000);
            jsonWill.put("funCode", 0);
            String strWillMsg = jsonWill.toString();
            connOpts.setWill("yg_scada_service_send_msg", strWillMsg.getBytes(), 0, false);

            tok = mqttClient.connect(connOpts);
            tok.waitForCompletion(-1);

            jsonErr.put("err", "ok");

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    private void connectCompleteCallback(boolean reconnect, String serverURI) throws MqttException, InterruptedException {
        log.info("---connectComplete, url=" + serverURI, " reconnect=" + reconnect);
        onStart();
        broadcastOperationMsg("MQTT服务器已经连接...");
        broadcastOnlineMsg();
    }

    private void connectionLostCallback(Throwable cause) {
        log.info("---connectionLost:" + cause.toString());
        cause.printStackTrace();
        broadcastOperationMsg("MQTT服务器连接中断...");
    }

    private void deliveryCompleteCallback(IMqttDeliveryToken token) {
        //log.info("---deliveryComplete");
    }

    private void messageArrivedCallback(String topic, MqttMessage message) {
    }


    public boolean publishMsgEx(String topic, JSONObject jsonMsg){
        boolean bRet = true;
        try {
            String strMsg = jsonMsg.toString();
            if (mqttClient != null) {
                mqttClient.publish(topic, strMsg.getBytes(), 0, false);
            }
        } catch (MqttException e) {
            log.info("publish_msg,topic:[" + topic + "] got an exception:" + e.toString());
            bRet = false;
        }
        return bRet;
    }

    public boolean publishMsg(String topic, JSONObject jsonMsg) {
        boolean bRet = true;
        int msgType = 0;
        if (jsonMsg.containsKey("msg_type")) {
            msgType = jsonMsg.getInteger("msg_type");
        } else if (jsonMsg.containsKey("webMsgType")) {
            msgType = jsonMsg.getInteger("webMsgType");
        }

        if (!paramHelper.isLogMsg()) {
            log.info("[SEND]-[" + topic + "] msgType=" + msgType);
        } else {
            log.info("[SEND]-[" + topic + "] " + jsonMsg.toJSONString());
        }

        try {
            String strMsg = jsonMsg.toString();
            if (mqttClient != null) {
                mqttClient.publish(topic, strMsg.getBytes(), 0, false);
            }
        } catch (MqttException e) {
            log.info("publish_msg,topic:[" + topic + "] got an exception:" + e.toString());
            bRet = false;
        }

        Date date = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");// 格式化日期
        String strDatetime = sdf.format(date);

        JSONObject jsonLogView = new JSONObject();
        jsonLogView.put("webMsgType", 1025);
        jsonLogView.put("funCode", (msgType < 1000)?1:2);
        jsonLogView.put("data", new JSONObject());
        jsonLogView.getJSONObject("data").put("datetime", strDatetime);
        jsonLogView.getJSONObject("data").put("target", (msgType < 1000)?"网关":"监控");
        jsonLogView.getJSONObject("data").put("direct", "接收");
        jsonLogView.getJSONObject("data").put("topic", topic);
        jsonLogView.getJSONObject("data").put("msgType", msgType);
        jsonLogView.getJSONObject("data").put("msg", paramHelper.isLogMsg()?jsonMsg:"不显示");

        publishMsgEx("yg_scada_service_send_msg", jsonLogView);
        return bRet;
    }

    public void broadcastOperationMsg(String msg) {
        Date date = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");// 格式化日期
        String strDatetime = sdf.format(date);

        JSONObject jsonMsg = new JSONObject();
        jsonMsg.put("webMsgType", 1000);
        jsonMsg.put("funCode", 2);
        jsonMsg.put("data", new JSONObject());
        jsonMsg.getJSONObject("data").put("msg", msg);
        jsonMsg.getJSONObject("data").put("time", strDatetime);
        String topic = "yg_scada_service_send_msg";

        if (!publishMsg(topic, jsonMsg)) {
            log.info("publish_msg,topic:[" + topic + "] got an exception:");
        }
    }

    public void broadcastOnlineMsg() {
        JSONObject jsonMsg = new JSONObject();
        jsonMsg.put("webMsgType", 1000);
        jsonMsg.put("funCode", 1);
        String topic = "yg_scada_service_send_msg";
        if (!publishMsg(topic, jsonMsg)) {
            log.info("publish_msg,topic:[" + topic + "] got an exception:");
        }
    }

    public void connectMqttServer() {
        broadcastOperationMsg("连接MQTT服务器....");
        start();
    }

    public void disconnectMqttServer() throws MqttException {
        broadcastOperationMsg("断开MQTT服务器....");
        stop();
    }

}
